/*
 * Copyright (c) 2024 Black Sesame Technologies
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "dal/allocator.h"
#include "dal/publisher.h"
#include "dal/config_loader.h"
#include "dal_fastdds/converter/StatusConverter.hpp"
#include "dal_fastdds/converter/StatusMaskConverter.hpp"
#include "dal_fastdds/converter/QosConverter.hpp"
#include "dal_fastdds/converter/RetTypeConverter.hpp"
#include "dal_fastdds/converter/InstanceHandleConverter.hpp"
#include "dal_fastdds/converter/GuidConverter.hpp"
#include "dal_fastdds/converter/WriteParamsConverter.hpp"
#include "dal_fastdds/listener_impl/PublisherListenerImpl.hpp"
#include "dal_fastdds/listener_impl/DataWriterListenerImpl.hpp"
#include "dal_fastdds/utils/log.hpp"
#include "dal_fastdds/utils/EntityGetter.hpp"
#include "dal_fastdds/utils/EntityMappingManager.hpp"
#include "dal_fastdds/config_manager/ConfigManager.hpp"
#ifdef DAL_ENABLE_DMABUFHEAP
#include "dal_fastdds/dbh/DalDbhBestEffortWriter.hpp"
#include "dal_fastdds/dbh/DalDbhReliableWriter.hpp"
#include "dal_dmabufheap/DmaBufHeapTopicSelector.hpp"
#include "dal_builtin/msg/dal_dma_buf_heap_send_notify_msg.hpp"
#endif
#include "dal_fastrtps_shared_cpp/TypeSupport.hpp"
#include "dalidl_typesupport_fastrtps_cpp/message_type_support.h"



using namespace eprosima::fastdds::dds;
using namespace eprosima::fastrtps::rtps;

extern const dal_instance_handle_t DAL_HANDLE_NIL;

// PUBLISHER
dal_publisher_listener_t*
dal_create_publisher_listener(dal_publisher_listener_callbacks_t callbacks)
{
    dal_publisher_listener_t* listener = dal_publisher_listener_allocate();
    if (listener == nullptr) {
        return nullptr;
    }
    listener->handle_ = new PublisherListenerImpl(callbacks);
    if (!listener->handle_) {
        LOG_ERROR() << "PublisherListenerImpl memory allocation failed." << LOG_ENDL;
        dal_publisher_listener_free(listener);
        return nullptr;
    }

    return listener;
}

dal_ret_t dal_delete_publisher_listener(dal_publisher_listener_t* listener)
{
    if (!listener || !listener->handle_) {
        LOG_ERROR_WITH_LOCATION() << "listener is invalid." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    delete reinterpret_cast<PublisherListenerImpl*>(listener->handle_);
    dal_publisher_listener_free(listener);
    reinterpret_cast<dal_publisher_listener_t*>(listener)->handle_ = nullptr;

    return DAL_RETCODE_OK;
}

dal_publisher_t* dal_domain_participant_create_publisher(dal_domain_participant_t* participant,
                                                         dal_publisher_qos_t* qos,
                                                         dal_publisher_listener_t* listener,
                                                         dal_status_mask_t mask)
{
    DomainParticipant* f_participant = get_fastdds_domain_participant(participant);
    if (!f_participant) {
        LOG_ERROR_WITH_LOCATION() << "get fastdds domain participant failed." << LOG_ENDL;
        return nullptr;
    }

    dal_publisher_t* d_pub = dal_publisher_allocate();
    if (!d_pub) {
        LOG_ERROR_WITH_LOCATION() << "dal publisher allocate failed." << LOG_ENDL;
        return nullptr;
    }

    PublisherQos f_qos;
    QosConverter::GetInstance()->ConvertPublisherQosFromDalToFastdds(qos, &f_qos);

    StatusMask f_mask;
    StatusMaskConverter::GetInstance()->ConvertStatusMaskFromDalToFastdds(mask, &f_mask);

    PublisherListenerImpl* f_listener =
        listener ? reinterpret_cast<PublisherListenerImpl*>(listener->handle_) : nullptr;

    auto f_pub = f_participant->create_publisher(f_qos, f_listener, f_mask);
    if (!f_pub) {
        LOG_ERROR_WITH_LOCATION() << "create publisher failed." << LOG_ENDL;
        dal_publisher_free(d_pub);
        return nullptr;
    }

    EntityMappingAddPublisher(f_pub, d_pub);

    DalPublisherImpl* handle_ = new DalPublisherImpl(f_pub, f_listener);
    handle_->SetFastddsDomainParticipant(f_participant);
    d_pub->handle_ = handle_;
    return d_pub;
}

dal_publisher_t*
dal_domain_participant_create_publisher_with_profile(dal_domain_participant_t* participant,
                                                     dal_publisher_qos_t* qos,
                                                     const char* profile_name,
                                                     dal_publisher_listener_t* listener,
                                                     dal_status_mask_t mask)
{
    if (!profile_name || strlen(profile_name) == 0) {
        return dal_domain_participant_create_publisher(participant, qos, listener, mask);
    }

    DomainParticipant* f_participant = get_fastdds_domain_participant(participant);
    if (!f_participant) {
        LOG_ERROR_WITH_LOCATION() << "get fastdds domain participant failed." << LOG_ENDL;
        return nullptr;
    }

    dal_publisher_t* d_pub = dal_publisher_allocate();
    if (!d_pub) {
        LOG_ERROR_WITH_LOCATION() << "dal publisher allocate failed." << LOG_ENDL;
        return nullptr;
    }

    PublisherQos f_qos;
    if (!ConfigManager::GetInstance()->getFastddsProfilePath().empty()) {
        try {
            f_participant->get_publisher_qos_from_profile(profile_name, f_qos);
        }
        catch (...) {
            LOG_ERROR_WITH_LOCATION()
                << "publisher qos profile [" << profile_name << "] in xml are not effective.";
            dal_publisher_free(d_pub);
            return nullptr;
        }
    } else {
        QosConverter::GetInstance()->ConvertPublisherQosFromDalToFastdds(qos, &f_qos);
    }

    StatusMask f_mask;
    StatusMaskConverter::GetInstance()->ConvertStatusMaskFromDalToFastdds(mask, &f_mask);

    PublisherListenerImpl* f_listener =
        listener ? reinterpret_cast<PublisherListenerImpl*>(listener->handle_) : nullptr;

    auto f_pub = f_participant->create_publisher(f_qos, f_listener, f_mask);
    if (!f_pub) {
        LOG_ERROR_WITH_LOCATION() << "create publisher failed." << LOG_ENDL;
        dal_publisher_free(d_pub);
        return nullptr;
    }

    EntityMappingAddPublisher(f_pub, d_pub);

    DalPublisherImpl* handle_ = new DalPublisherImpl(f_pub, f_listener);
    handle_->SetFastddsDomainParticipant(f_participant);
    d_pub->handle_ = handle_;
    return d_pub;
}

dal_ret_t dal_domain_participant_delete_publisher(dal_domain_participant_t* participant,
                                                  dal_publisher_t* pub)
{
    DomainParticipant* f_participant = get_fastdds_domain_participant(participant);
    if (!f_participant) {
        LOG_ERROR_WITH_LOCATION() << "get fastdds domain participant failed." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    Publisher* f_pub = get_fastdds_publisher(pub);
    if (!f_pub) {
        LOG_ERROR_WITH_LOCATION() << "get fastdds publisher failed." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    ReturnCode_t f_ret = f_participant->delete_publisher(f_pub);
    dal_ret_t d_ret = RetTypeConverter::GetInstance()->ConvertRetFromFastddsToDal(f_ret);
    if (DAL_RETCODE_OK != d_ret) {
        LOG_ERROR_WITH_LOCATION() << "delete publisher failed." << LOG_ENDL;
        return d_ret;
    }

    EntityMappingRemovePublisher(f_pub);

    DalPublisherImpl* pub_impl = reinterpret_cast<DalPublisherImpl*>(pub->handle_);
    delete pub_impl;
    dal_publisher_free(pub);
    reinterpret_cast<dal_publisher_t*>(pub)->handle_ = nullptr;

    return DAL_RETCODE_OK;
}

// DATAWRITER
dal_datawriter_listener_t*
dal_create_datawriter_listener(dal_datawriter_listener_callbacks_t callbacks)
{
    dal_datawriter_listener_t* listener = dal_datawriter_listener_allocate();
    if (listener == nullptr) {
        return nullptr;
    }
    listener->handle_ = new DataWriterListenerImpl(callbacks);
    if (!listener->handle_) {
        LOG_ERROR() << "DataWriterListenerImpl memory allocation failed." << LOG_ENDL;
        dal_datawriter_listener_free(listener);
        return nullptr;
    }

    return listener;
}

dal_ret_t dal_delete_datawriter_listener(dal_datawriter_listener_t* listener)
{
    if (!listener || !listener->handle_) {
        LOG_ERROR_WITH_LOCATION() << "listener is invalid." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    delete reinterpret_cast<DataWriterListenerImpl*>(listener->handle_);
    dal_datawriter_listener_free(listener);
    reinterpret_cast<dal_datawriter_listener_t*>(listener)->handle_ = nullptr;

    return DAL_RETCODE_OK;
}

static bool DalCreateDataWriterParaCheck(dal_publisher_t* pub, dal_topic_t* topic)
{
    if (!pub || !pub->handle_) {
        LOG_ERROR_WITH_LOCATION() << "publisher is nullptr." << LOG_ENDL;
        return false;
    }

    Publisher* f_pub = get_fastdds_publisher(pub);
    if (!f_pub) {
        LOG_ERROR_WITH_LOCATION() << "get fastdds publisher failed." << LOG_ENDL;
        return false;
    }

    Topic* f_topic = get_fastdds_topic(topic);
    if (!f_topic) {
        LOG_ERROR_WITH_LOCATION() << "get fastdds topic failed." << LOG_ENDL;
        return false;
    }

    if (!topic->type_support_impl_) {
        LOG_ERROR_WITH_LOCATION() << "topic type support is nullptr." << LOG_ENDL;
        return false;
    }

    return true;
}

static dal_datawriter_t* DalCreateDataWriterByTopicType(dal_publisher_t* pub,
                                                        dal_topic_t* topic,
                                                        const DataWriterQos& f_qos,
                                                        DataWriterListenerImpl* f_listener,
                                                        const StatusMask& f_mask)
{
    Publisher* f_pub = get_fastdds_publisher(pub);
    Topic* f_topic = get_fastdds_topic(topic);
    std::string topic_name = f_topic->get_name();
    dal_datawriter_t* d_writer = dal_datawriter_allocate();
    if (!d_writer) {
        LOG_ERROR_WITH_LOCATION() << "dal datawriter allocate failed." << LOG_ENDL;
        return nullptr;
    }
    d_writer->type_support_impl_ = topic->type_support_impl_;
#ifdef DAL_ENABLE_DMABUFHEAP
    using namespace dal_dmabufheap;
    DmaBufHeapTopicSelector* topic_selector = DmaBufHeapTopicSelector::GetInstance();

    if (topic_selector->IsDbhTopic(topic_name)) {
        DalDataWriterImpl* writer_handle_ = new DalDataWriterImpl(nullptr, nullptr);
        DalPublisherImpl* pub_handle_ = reinterpret_cast<DalPublisherImpl*>(pub->handle_);

        if (f_qos.reliability().kind == BEST_EFFORT_RELIABILITY_QOS) {
            writer_handle_->dbh_writer_ = new DalDbhBestEffortWriter();
        } else {
            writer_handle_->dbh_writer_ = new DalDbhReliableWriter();
        }

        writer_handle_->dbh_writer_->SetFastddsDomainParticipant(
            pub_handle_->FastddsDomainParticipant());
        writer_handle_->dbh_writer_->SetFastddsPublisher(f_pub);
        writer_handle_->dbh_writer_->SetDbhSendNotifyWriterListener(f_listener);
        writer_handle_->dbh_writer_->SetDbhSendNotifyWriterQos(f_qos);
        writer_handle_->dbh_writer_->SetDbhSendNotifyWriterStatusMask(f_mask);
        writer_handle_->dbh_writer_->SetTypeSupport(d_writer->type_support_impl_);
        writer_handle_->dbh_writer_->SetOriginTopicName(topic_name);
        if (!writer_handle_->dbh_writer_->CreateDbhBuiltinEntities()) {
            LOG_ERROR_WITH_LOCATION() << "create dbh builtin entities failed." << LOG_ENDL;
            delete writer_handle_;
            dal_datawriter_free(d_writer);
            return nullptr;
        }

        // d_writer could be created in on_publication_matched, check it first.
        // EntityMappingAddDataWriter(f_writer, d_writer);
        dal_datawriter_t* d_writer_tmp = EntityMappingGetOrAddDataWriter(
            writer_handle_->dbh_writer_->GetDbhSendNotifyWriter(), d_writer);
        if (d_writer_tmp != d_writer) {
            dal_datawriter_free(d_writer);
            d_writer_tmp->type_support_impl_ = topic->type_support_impl_;
            d_writer = d_writer_tmp;
        }

        writer_handle_->SetFastddsDomainParticipant(pub_handle_->FastddsDomainParticipant());
        writer_handle_->SetFastddsPublisher(f_pub);
        writer_handle_->type_ = DalDataWriterImpl::WriterType::DBH_WRITER;
        d_writer->handle_ = writer_handle_;
        return d_writer;
    } else if (topic_selector->IsMixedTopic(f_topic->get_name())) {
        LOG_ERROR_WITH_LOCATION() << "mixed topic is not supported yet." << LOG_ENDL;
        dal_datawriter_free(d_writer);
        return nullptr;
    } else {
#endif
        auto f_writer = f_pub->create_datawriter(f_topic, f_qos, f_listener, f_mask);
        if (!f_writer) {
            LOG_ERROR_WITH_LOCATION() << "create datawriter failed." << LOG_ENDL;
            dal_datawriter_free(d_writer);
            return nullptr;
        }

        // d_writer could be created in on_publication_matched, check it first.
        // EntityMappingAddDataWriter(f_writer, d_writer);
        dal_datawriter_t* d_writer_tmp = EntityMappingGetOrAddDataWriter(f_writer, d_writer);
        if (d_writer_tmp != d_writer) {
            dal_datawriter_free(d_writer);
            d_writer_tmp->type_support_impl_ = topic->type_support_impl_;
            d_writer = d_writer_tmp;
        }

        DalDataWriterImpl* writer_handle_ = new DalDataWriterImpl(f_writer, f_listener);
        DalPublisherImpl* pub_handle_ = reinterpret_cast<DalPublisherImpl*>(pub->handle_);
        writer_handle_->SetFastddsDomainParticipant(pub_handle_->FastddsDomainParticipant());
        writer_handle_->SetFastddsPublisher(f_pub);
        writer_handle_->SetTypeSupport(d_writer->type_support_impl_);
        writer_handle_->type_ = DalDataWriterImpl::WriterType::DDS_WRITER;
        d_writer->handle_ = writer_handle_;

        return d_writer;
#ifdef DAL_ENABLE_DMABUFHEAP
    }
#endif
}

dal_datawriter_t* dal_publisher_create_datawriter(dal_publisher_t* pub,
                                                  dal_topic_t* topic,
                                                  dal_datawriter_qos_t* qos,
                                                  dal_datawriter_listener_t* listener,
                                                  dal_status_mask_t mask)
{
    if (!DalCreateDataWriterParaCheck(pub, topic)) {
        LOG_ERROR_WITH_LOCATION() << "DalCreateDataWriterParaCheck failed." << LOG_ENDL;
        return nullptr;
    }

    DataWriterQos f_qos;
    QosConverter::GetInstance()->ConvertDataWriterQosFromDalToFastdds(qos, &f_qos);
    f_qos.endpoint().history_memory_policy =
        eprosima::fastrtps::rtps::MemoryManagementPolicy::PREALLOCATED_WITH_REALLOC_MEMORY_MODE;

#ifdef DAL_FASTDDS_DATA_SHARING_DISABLED
    f_qos.data_sharing().off();
#endif

    StatusMask f_mask;
    StatusMaskConverter::GetInstance()->ConvertStatusMaskFromDalToFastdds(mask, &f_mask);

    DataWriterListenerImpl* f_listener =
        listener ? reinterpret_cast<DataWriterListenerImpl*>(listener->handle_) : nullptr;

    return DalCreateDataWriterByTopicType(pub, topic, f_qos, f_listener, f_mask);
}

dal_datawriter_t* dal_publisher_create_datawriter_with_profile(dal_publisher_t* pub,
                                                               dal_topic_t* topic,
                                                               dal_datawriter_qos_t* qos,
                                                               const char* profile_name,
                                                               dal_datawriter_listener_t* listener,
                                                               dal_status_mask_t mask)
{
    if (!profile_name || strlen(profile_name) == 0) {
        return dal_publisher_create_datawriter(pub, topic, qos, listener, mask);
    }

    if (!DalCreateDataWriterParaCheck(pub, topic)) {
        LOG_ERROR_WITH_LOCATION() << "DalCreateDataWriterParaCheck failed." << LOG_ENDL;
        return nullptr;
    }

    DataWriterQos f_qos;
    if (!ConfigManager::GetInstance()->getFastddsProfilePath().empty()) {
        try {
            Publisher* f_pub = get_fastdds_publisher(pub);
            f_pub->get_datawriter_qos_from_profile(profile_name, f_qos);
        }
        catch (...) {
            LOG_ERROR_WITH_LOCATION() << "load xml failed, datawriter qos profile[" << profile_name
                                      << "] in xml are not effective.";
            return nullptr;
        }
    } else {
        QosConverter::GetInstance()->ConvertDataWriterQosFromDalToFastdds(qos, &f_qos);
    }
    f_qos.endpoint().history_memory_policy =
        eprosima::fastrtps::rtps::MemoryManagementPolicy::PREALLOCATED_WITH_REALLOC_MEMORY_MODE;

#ifdef DAL_FASTDDS_DATA_SHARING_DISABLED
    f_qos.data_sharing().off();
#endif

    StatusMask f_mask;
    StatusMaskConverter::GetInstance()->ConvertStatusMaskFromDalToFastdds(mask, &f_mask);

    DataWriterListenerImpl* f_listener =
        listener ? reinterpret_cast<DataWriterListenerImpl*>(listener->handle_) : nullptr;

    return DalCreateDataWriterByTopicType(pub, topic, f_qos, f_listener, f_mask);
}

static bool DalDeleteDataWriterParaCheck(dal_publisher_t* pub, dal_datawriter_t* writer)
{
    if (!pub || !pub->handle_) {
        LOG_ERROR_WITH_LOCATION() << "param pub invalid." << LOG_ENDL;
        return false;
    }

    if (!writer || !writer->handle_) {
        LOG_ERROR_WITH_LOCATION() << "param writer invalid." << LOG_ENDL;
        return false;
    }

    return true;
}

dal_ret_t dal_publisher_delete_datawriter(dal_publisher_t* pub, dal_datawriter_t* writer)
{
    if (!DalDeleteDataWriterParaCheck(pub, writer)) {
        LOG_ERROR_WITH_LOCATION() << "param check failed." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    DalDataWriterImpl* writer_impl = reinterpret_cast<DalDataWriterImpl*>(writer->handle_);
#ifdef DAL_ENABLE_DMABUFHEAP
    if (writer_impl->type_ == DalDataWriterImpl::WriterType::DDS_WRITER) {
#endif
        Publisher* f_pub = get_fastdds_publisher(pub);
        if (!f_pub) {
            LOG_ERROR_WITH_LOCATION() << "get fastdds publisher failed." << LOG_ENDL;
            return DAL_RETCODE_BAD_PARAMETER;
        }

        DataWriter* f_writer = get_fastdds_datawriter(writer);
        if (!f_writer) {
            LOG_ERROR_WITH_LOCATION() << "get fastdds datawriter failed." << LOG_ENDL;
            return DAL_RETCODE_BAD_PARAMETER;
        }

        ReturnCode_t f_ret = f_pub->delete_datawriter(f_writer);
        dal_ret_t d_ret = RetTypeConverter::GetInstance()->ConvertRetFromFastddsToDal(f_ret);
        if (DAL_RETCODE_OK != d_ret) {
            LOG_ERROR_WITH_LOCATION() << "delete publisher failed." << LOG_ENDL;
            return d_ret;
        }

        EntityMappingRemoveDataWriter(f_writer);
#ifdef DAL_ENABLE_DMABUFHEAP
    } else if (writer_impl->type_ == DalDataWriterImpl::WriterType::DBH_WRITER) {
        DataWriter* f_writer = writer_impl->dbh_writer_->GetDbhSendNotifyWriter();
        EntityMappingRemoveDataWriter(f_writer);

    } else {
        std::cout << "mixed topic is not supported yet." << std::endl;
    }
#endif
    delete writer_impl;
    dal_datawriter_free(writer);
    reinterpret_cast<dal_datawriter_t*>(writer)->handle_ = nullptr;

    return DAL_RETCODE_OK;
}

static bool DalWriteParaCheck(dal_datawriter_t* writer, dal_data data)
{
    if (!writer || !writer->handle_) {
        LOG_ERROR_WITH_LOCATION() << "param writer invalid." << LOG_ENDL;
        return false;
    }

    if (!data) {
        LOG_ERROR_WITH_LOCATION() << "param data invalid." << LOG_ENDL;
        return false;
    }

    return true;
}

dal_ret_t dal_datawriter_write(dal_datawriter_t* writer,
                               dal_data data,
                               dal_instance_handle_t handle = DAL_HANDLE_NIL)
{
    if (!DalWriteParaCheck(writer, data)) {
        LOG_ERROR_WITH_LOCATION() << "param check failed." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    InstanceHandle_t f_handle;
    dal_ret_t d_ret = InstanceHandleConverter::GetInstance()->ConvertInstanceHandleFromDalToFastdds(
        &handle, &f_handle);
    if (DAL_RETCODE_OK != d_ret)
        return d_ret;

    DalDataWriterImpl* writer_impl = reinterpret_cast<DalDataWriterImpl*>(writer->handle_);
#ifdef DAL_ENABLE_DMABUFHEAP
    if (writer_impl->type_ == DalDataWriterImpl::WriterType::DDS_WRITER) {
#endif
        DataWriter* f_writer = get_fastdds_datawriter(writer);
        if (!f_writer) {
            LOG_ERROR_WITH_LOCATION() << "get fastdds datawriter failed." << LOG_ENDL;
            return DAL_RETCODE_BAD_PARAMETER;
        }

        ReturnCode_t f_ret = writer_impl->DDSWrite(data, f_handle);
        return RetTypeConverter::GetInstance()->ConvertRetFromFastddsToDal(f_ret);
#ifdef DAL_ENABLE_DMABUFHEAP
    } else if (writer_impl->type_ == DalDataWriterImpl::WriterType::DBH_WRITER) {
        if (!writer_impl->dbh_writer_) {
            LOG_ERROR_WITH_LOCATION() << "dbh_writer is nullptr." << LOG_ENDL;
            return DAL_RETCODE_ERROR;
        }

        ReturnCode_t f_ret = writer_impl->DBHWrite(data, f_handle);
        return RetTypeConverter::GetInstance()->ConvertRetFromFastddsToDal(f_ret);
    } else {
        std::cout << "mixed topic is not supported yet." << std::endl;
    }

    return DAL_RETCODE_ERROR;
#endif
}

dal_ret_t dal_datawriter_write_with_params(dal_datawriter_t* writer,
                                           dal_data data,
                                           dal_write_params_t* write_params)
{
    WriteParams f_params;
    ReturnCode_t f_ret = DAL_RETCODE_OK;

    DataWriter* f_writer = get_fastdds_datawriter(writer);
    if (!f_writer) {
        LOG_ERROR_WITH_LOCATION() << "get fastdds datawriter failed." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }
    if (!write_params) {
        LOG_ERROR_WITH_LOCATION() << "write params is nullptr." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    WriteParamsConverter::GetInstance()->ConvertWriteParamsFromDalToFastdds(write_params,
                                                                            &f_params);
    dal_fastrtps_shared_cpp::SerializedData sdata;
    sdata.is_cdr_buffer = false;
    sdata.data = static_cast<void*>(data);
    sdata.impl = writer->type_support_impl_;

    if (!f_writer->write(&sdata, f_params)) {
        f_ret = DAL_RETCODE_ERROR;
    }
    return RetTypeConverter::GetInstance()->ConvertRetFromFastddsToDal(f_ret);
}

dal_guid_t dal_datawriter_get_guid(dal_datawriter_t* writer)
{
    dal_guid_t d_guid = { };
    DataWriter* f_writer = get_fastdds_datawriter(writer);
    if (!f_writer) {
        LOG_ERROR_WITH_LOCATION() << "get fastdds datawriter failed." << LOG_ENDL;
        return d_guid;
    }

    auto f_guid = f_writer->guid();
    GuidConverter::GetInstance()->ConvertGuidFromFastddsToDal(&f_guid, &d_guid);
    return d_guid;
}
